package com.hs.kfk.basic;

import org.apache.kafka.clients.consumer.*;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Description: 基础班的消息消费方
 * @Author 胡尚
 * @Date: 2024/8/7 19:07
 */
public class BasicConsumer {
    /**
     * 定义kafka服务端地址
     */
    // private final static String BOOTSTRAP_SERVER = "192.168.75.61:9092,192.168.75.62:9092,192.168.75.63:9092";
    private final static String BOOTSTRAP_SERVER = "worker1:9092,worker2:9092,worker3:9092";
    /**
     * 生产者往哪个topic中发送消息
     */
    private final static String TOPIC_NAME = "disTopic";
    /**
     * 消费者组名
     */
    private final static String CONSUMER_GROUP = "test";

    public static void main(String[] args) {
        // 设置消费者相关的属性
        Properties properties = new Properties();
        // 设置kafka端口
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        // 每个消费者需要指定一个消费者组
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP);
        // 配置key的序列化类 去找org.apache.kafka.common.serialization.Deserializer接口的实现类
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 配置value的序列化类
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 创建一个消息消费者对象
        // 这里的key和value的泛型要和上方定义的序列化类型匹配上
        Consumer<String, String> consumer = new KafkaConsumer<>(properties);
        // 消费者可以订阅多个topic
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            // 消费者拉取消息  100毫秒超时时间
            // 这里一次拉取的是一批消息
            // 这里的key和value的泛型要和上方定义的序列化类型匹配上
            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(100));

            // 处理消息
            for (ConsumerRecord<String, String> record : consumerRecords) {
                int partition = record.partition();
                long offset = record.offset();
                String topic = record.topic();
                String key = record.key();
                String message = record.value();
                System.out.println("topic:" + topic + "\tpartition:" + partition + "\toffset: " + offset + "\tkey: " + key + "\tmessage: " + message);
            }

            // 提交offset消息就不会重复消费
            //同步提交，表示必须等到offset提交完毕，再去消费下一批数据。
            consumer.commitSync();
            // 异步提交，表示发送完提交offset请求后，就开始消费下一批数据了。不用等到Broker的确认。
            // consumer.commitAsync();

        }
    }
}
